﻿#include "listen_thread.h"
#include "tb_interface.h"
#include "../io/pcapio.h"
listen_thread::listen_thread(
		const TASKBUS::cmdlineParser * cmdline,
		QObject * parent)
	:QThread(parent)
	,m_cmd(cmdline)
{
	m_total_pack = 0;
}

QAtomicInteger<quint64> listen_thread::m_refTms = 0;

void listen_thread::run()
{
	using namespace TASKBUS;
	bool bfinished = false;
	const int sub_in = m_cmd->toInt("packin",0);
	emit sig_message("STDIN thread started.");
	while (false==bfinished)
	{
		subject_package_header header;
		std::vector<unsigned char> packagedta = pull_subject(&header);
		if (!is_valid_header(header))
		{
			fprintf(stderr,"Recived BAD Command.");
			fflush(stderr);
			msleep(100);
			continue;
		}
		if (!packagedta.empty())
		{
			if ( is_control_subject(header))
			{
				//收到命令进程退出的广播消息,退出
				if (strstr((const char *)packagedta.data(),"function=quit;")!=nullptr)
				{
					fprintf(stderr,"Recived Quit Command.");
					fflush(stderr);
					bfinished = true;
				}
			}
			else if (header.subject_id==sub_in && sub_in>0)
			{
				if (PCAPIO::enqueuePack((const char *)packagedta.data(),
										header.data_length,TBTUNID,TBPORTNAME))
				{
					if (++m_total_pack % 1000==0)
						emit sig_message(tr("TaskBus recv %1").arg(m_total_pack));
				}
			}
		}
	}
	emit quit_app();
}

void stdoutLoop(TASKBUS::cmdlineParser * args, std::function<void(QString)> msg)
{
	const int argOut = args->toInt("packout",0);
	const int inst = args->toInt("instance",0);
	unsigned long long TPos = 0;
	msg("STDOUT Thread started.");
	while (argOut>0)
	{
		quint64 rp = PCAPIO::pcap_recv_pos;
		if (TPos + PCAPIO_BUFCNT < rp)
		{
			msg("Stdout tunnel overflow, clear.");
			TPos = rp-1;
		}
		while (TPos < rp) {
			const int nPOS = TPos % PCAPIO_BUFCNT;
			const int fromID = PCAPIO::global_buffer[nPOS].from_id;
			if (fromID !=TBTUNID
					&& PCAPIO::global_buffer[nPOS].len < 65536
					&& PCAPIO::global_buffer[nPOS].len >0)
			{
				const unsigned short len = PCAPIO::global_buffer[nPOS].len;
				TASKBUS::push_subject(argOut,inst,len,
									  (const unsigned char *)
									  PCAPIO::global_buffer[nPOS].data.constData());
			}
			++TPos;
		}
		QThread::msleep(10);
	}
	msg("STDOUT Thread quit.");
}
